package com.txby.dplus.core.util;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.ObjectId;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.filerep.KettleFileRepository;
import org.pentaho.di.repository.filerep.KettleFileRepositoryMeta;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;  
/**
 * kettle 的转换和作业执行操作
* @ClassName: KettleExecuUtil 
* @Description: TODO(这里用一句话描述这个类的作用) 
* @author jacking
* @date 2018年1月17日 下午12:10:07 
*
 */
public class KettleExecuUtil {  
	/**
	 * DB资源库 
	 */
	private static KettleDatabaseRepository dbrepository;
	/**
	 * 文件目录库
	 */
	private static KettleFileRepository filerepository;

	 private static KettleExecuUtil keu=null;
	
    private KettleExecuUtil(){
    	try {
			//初始化
			KettleEnvironment.init();
			EnvUtil.environmentInit(); 
			//创建DB资源库 
			dbrepository=new KettleDatabaseRepository(); 

			//文件目录库
			filerepository = new KettleFileRepository();
		} catch (KettleException e) {
			e.printStackTrace();
		} 
    	
    }
    public  static DatabaseMeta getDatabaseMeta(String host,String port,String dbName,String user,String pass){
    	DatabaseMeta databaseMeta=null;
    	try {
			//初始化
			KettleEnvironment.init();
			EnvUtil.environmentInit(); 
			databaseMeta=new DatabaseMeta("kettle","mysql","jdbc",host,dbName,port,user,pass);
		} catch (KettleException e) {
			e.printStackTrace();
		} 
        return databaseMeta;
    } 
    public  static DatabaseMeta getDatabaseMeta(String name,String host,String type,String port,String dbName,String user,String pass){
    	DatabaseMeta databaseMeta=null;
    	try {
    		//初始化
    		KettleEnvironment.init();
    		EnvUtil.environmentInit(); 
    		databaseMeta=new DatabaseMeta(name,type,"jdbc",host,dbName,port,user,pass);
    	} catch (KettleException e) {
    		e.printStackTrace();
    	} 
    	return databaseMeta;
    } 
    public synchronized static KettleExecuUtil getInstance(){
    	if(keu == null){
    		keu= new KettleExecuUtil();
    	}
    	return keu;
    } 
	/**
	 * 环境初始化
	* @Title: init 
	* @Description: TODO(这里用一句话描述这个方法的作用) 
	* @param     设定文件 
	* @return void    返回类型 
	* @throws
	 */
	  public static boolean init(){
		try {
			//初始化
			KettleEnvironment.init();
			EnvUtil.environmentInit(); 
			//创建DB资源库 
			dbrepository=new KettleDatabaseRepository(); 
		} catch (KettleException e) {
			e.printStackTrace();
		} 
		return true;
		
	}
  /**
   * 调用kjb转换文件
  * @Title: runJob 
  * @Description: TODO(这里用一句话描述这个方法的作用) 
  * @param @param jobname    设定文件 
  * @return void    返回类型 
  * @throws
   */
    public   void runJob(String jobPath) {  
        try {  
            // jobname 是Job脚本的路径及名称  
            JobMeta jobMeta = new JobMeta(jobPath, null);  
            Job job = new Job(null, jobMeta);  
            // 向Job 脚本传递参数，脚本中获取参数值：${参数名}  
            // job.setVariable(paraname, paravalue);  
            job.start();  
            job.waitUntilFinished();  
            if (job.getErrors() > 0) {  
            	 throw new Exception("执行job发生异常");  
            }  
        } catch (Exception e) {  
        	 e.printStackTrace();  
        }  
    }  
  
    /**
     * 调用ktr转换文件
    * @Title: runTrans 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param filename    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public   void runTrans(String ktrPath) {  
        try {  
            TransMeta transMeta = new TransMeta(ktrPath);  
            Trans trans = new Trans(transMeta);  
            trans.prepareExecution(null);  
            trans.startThreads();  
            trans.waitUntilFinished();  
            if (trans.getErrors() != 0) {  
            	 throw new Exception("执行trans发生异常");  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
    /**
     * 调用ktr转换文件,同时传递参数
    * @Title: runTrans 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param params
    * @param @param ktrPath    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public   void runTrans(String[] params, String ktrPath) {  
        Trans trans = null;  
        try {  
            TransMeta transMeta = new TransMeta(ktrPath);  
            // 转换  
            trans = new Trans(transMeta);  
            // 执行转换  
            trans.execute(params);  
            // 等待转换执行结束  
            trans.waitUntilFinished();  
            // 抛出异常  
            if (trans.getErrors() > 0) {  
            	 throw new Exception("执行trans发生异常");  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    } 
    /**
     * 调用ktr转换文件,同时传递参数
     * @Title: runTrans 
     * @Description: TODO(这里用一句话描述这个方法的作用) 
     * @param @param params
     * @param @param ktrPath    设定文件 
     * @return void    返回类型 
     * @throws
     */
    public   void runTrans(Map<String,String> maps, String ktrPath) {  
    	Trans trans = null;  
    	try {  
    		TransMeta transMeta = new TransMeta(ktrPath);  
    		// 转换  
    		trans = new Trans(transMeta);  
    		if(maps!=null){
    			for(Entry<String, String> entry:maps.entrySet()){
    				String key=entry.getKey();
    				String value=entry.getValue();
    				trans.setVariable(key, value);
    			}
    		}
    		 trans.execute(null);
    		// 等待转换执行结束  
    		trans.waitUntilFinished();  
    		// 抛出异常  
    		if (trans.getErrors() > 0) {  
    			throw new Exception("执行trans发生异常");  
    		}  
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    } 
    /**
     * 根据输入流执行转换，同时传递参数
    * @Title: runTrans 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param maps
    * @param @param in    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public   void runTrans(Map<String,String> maps, InputStream in ){ 
    	Trans trans = null;  
    	try {  
    		TransMeta transMeta = new TransMeta(in, null, false, null, null) ;
    		// 转换  
    		trans = new Trans(transMeta);  
    		if(maps!=null){
    			for(Entry<String, String> entry:maps.entrySet()){
    				String key=entry.getKey();
    				String value=entry.getValue();
    				trans.setVariable(key, value);
    			}
    		}
    		trans.execute(null);
    		// 等待转换执行结束  
    		trans.waitUntilFinished();  
    		// 抛出异常  
    		if (trans.getErrors() > 0) {  
    			throw new Exception("执行trans发生异常");  
    		}  
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    } 
    /**
     * 根据输入流获取xml TransMeta
    * @Title: getTransMeta 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param in
    * @param @return    设定文件 
    * @return TransMeta    返回类型 
    * @throws
     */
    public   TransMeta getTransMeta( InputStream in){ 
    	TransMeta transMeta = null;  
    	try {  
    		 transMeta = new TransMeta(in, null, false, null, null) ;
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    	return transMeta;
    } 
    /**
     * 根据xml字符获取xml TransMeta
    * @Title: getTransMeta 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param text
    * @param @return    设定文件 
    * @return TransMeta    返回类型 
    * @throws
     */
    public   TransMeta getTransMeta(String text){ 
    	TransMeta transMeta = null;  
    	try {  
    		transMeta =new TransMeta(getStringStream(text), null, false, null, null) ;
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    	return transMeta;
    } 
    /**
     * 根据输入流获取xml JobMeta
    * @Title: getJobMeta 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param in
    * @param @return    设定文件 
    * @return JobMeta    返回类型 
    * @throws
     */
    public   JobMeta getJobMeta( InputStream in){ 
    	JobMeta jobMeta = null;  
    	try {  
    		jobMeta =  new JobMeta(in, null, null);
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    	return jobMeta;
    } 
   /**
    *  根据xml字符获取xml JobMeta
   * @Title: getJobMeta 
   * @Description: TODO(这里用一句话描述这个方法的作用) 
   * @param @param text
   * @param @return    设定文件 
   * @return JobMeta    返回类型 
   * @throws
    */
    public   JobMeta getJobMeta(String text){ 
    	JobMeta jobMeta = null;  
    	try {  
    		  jobMeta = new JobMeta(getStringStream(text), null, null);
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    	return jobMeta;
    } 
    /**
     * 根据输完整XML执行转换，同时传递参数
     * @Title: runTrans 
     * @Description: TODO(这里用一句话描述这个方法的作用) 
     * @param @param maps 同时传递参数
     * @param @param text    字符XML
     * @return void    返回类型 
     * @throws
     */
    public   void runTransByString(Map<String,String> maps, String text ){ 
    	Trans trans = null;  
    	try {  
    		TransMeta transMeta = new TransMeta(getStringStream(text), null, false, null, null) ;
    		// 转换  
    		trans = new Trans(transMeta);  
    		if(maps!=null){
    			for(Entry<String, String> entry:maps.entrySet()){
    				String key=entry.getKey();
    				String value=entry.getValue();
    				trans.setVariable(key, value);
    			}
    		}
    		trans.execute(null);
    		// 等待转换执行结束  
    		trans.waitUntilFinished();  
    		// 抛出异常  
    		if (trans.getErrors() > 0) {  
    			throw new Exception("执行trans发生异常");  
    		}  
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    } 
    /**
     * 根据输入流执行job,同时传递参数
    * @Title: runJob 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param maps
    * @param @param jobPath    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public   void runJob(Map<String,String> maps, InputStream in ) {  
        try {  
            // jobname 是Job脚本的路径及名称  
            JobMeta jobMeta = new JobMeta(in, null, null);
            Job job = new Job(null, jobMeta);  
            // 向Job 脚本传递参数，脚本中获取参数值：${参数名}  
            // job.setVariable(paraname, paravalue);  
            Set<Entry<String, String>> set=maps.entrySet(); 
            for(Iterator<Entry<String, String>> it=set.iterator();it.hasNext();){ 
                Entry<String, String> ent=it.next(); 
                job.setVariable(ent.getKey(), ent.getValue());  
            } 
            job.start();  
            job.waitUntilFinished();  
            if (job.getErrors() > 0) {  
                throw new Exception("执行job发生异常");  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    } 
    /**
     * 根据完整XML执行job,同时传递参数
     * @Title: runJob 
     * @Description: TODO(这里用一句话描述这个方法的作用) 
     * @param @param maps
     * @param @param text    字符XML
     * @return void    返回类型 
     * @throws
     */
    public   void runJobByString(Map<String,String> maps, String text ) {  
    	try {  
    		// jobname 是Job脚本的路径及名称  
    		JobMeta jobMeta = new JobMeta(getStringStream(text), null, null);
    		Job job = new Job(null, jobMeta);  
    		// 向Job 脚本传递参数，脚本中获取参数值：${参数名}  
    		// job.setVariable(paraname, paravalue);  
    		Set<Entry<String, String>> set=maps.entrySet(); 
    		for(Iterator<Entry<String, String>> it=set.iterator();it.hasNext();){ 
    			Entry<String, String> ent=it.next(); 
    			job.setVariable(ent.getKey(), ent.getValue());  
    		} 
    		job.start();  
    		job.waitUntilFinished();  
    		if (job.getErrors() > 0) {  
    			throw new Exception("执行job发生异常");  
    		}  
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    } 
    /**
     * 调用kjb转换文件,同时传递参数
    * @Title: runJob 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param maps
    * @param @param jobPath    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public   void runJob(Map<String,String> maps, String jobPath) {  
    	try {  
    		// jobname 是Job脚本的路径及名称  
    		JobMeta jobMeta = new JobMeta(jobPath, null);  
    		Job job = new Job(null, jobMeta);  
    		// 向Job 脚本传递参数，脚本中获取参数值：${参数名}  
    		// job.setVariable(paraname, paravalue);  
    		Set<Entry<String, String>> set=maps.entrySet(); 
    		for(Iterator<Entry<String, String>> it=set.iterator();it.hasNext();){ 
    			Entry<String, String> ent=it.next(); 
    			job.setVariable(ent.getKey(), ent.getValue());  
    		} 
    		job.start();  
    		job.waitUntilFinished();  
    		if (job.getErrors() > 0) {  
    			throw new Exception("执行job发生异常");  
    		}  
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    } 
    /**
     * demo调用资源库执行转换
    * @Title: runWithDbDemo 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param     设定文件 
    * @return void    返回类型 
    * @throws
     */
    public   void runTransWithDbDemo()  { 
        try {
			DatabaseMeta databaseMeta=new DatabaseMeta("kettle","mysql","jdbc","192.168.2.71","kettle","3306","user","123456"); 
			//选择资源库 
			KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("kettle","kettle","Transformation description",databaseMeta); 
			dbrepository.init(kettleDatabaseRepositoryMeta); 
			//连接资源库 
			dbrepository.connect("admin","admin"); 
			RepositoryDirectoryInterface directoryInterface=dbrepository.loadRepositoryDirectoryTree(); 
			//选择转换 
			TransMeta transMeta=dbrepository.loadTransformation("测试数据异构同步和更新",directoryInterface,null,true,null); 
			Trans trans=new Trans(transMeta); 
			trans.execute(null); 
			trans.waitUntilFinished();//等待直到数据结束 
			if(trans.getErrors()>0){ 
			    System.out.println("transformation error"); 
			}else{ 
			    System.out.println("transformation successfully"); 
			}
		} catch (KettleException e) {
			e.printStackTrace();
		} 
   } 
    public   void runTransWithDbDemo2()  { 
    	try {
    		DatabaseMeta databaseMeta=new DatabaseMeta("kettle","mysql","jdbc","192.168.2.71","kettle","3306","user","123456"); 
    		//选择资源库 
    		KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("kettle","kettle","Transformation description",databaseMeta); 
    		dbrepository.init(kettleDatabaseRepositoryMeta); 
    		//连接资源库 
    		dbrepository.connect("admin","admin"); 
    		RepositoryDirectoryInterface directoryInterface=dbrepository.loadRepositoryDirectoryTree(); 
    		//选择转换 
    		TransMeta transMeta=dbrepository.loadTransformation("测试数据异构同步和更新",directoryInterface,null,true,null); 
    		Trans trans=new Trans(transMeta); 
    		trans.execute(null); 
    		trans.waitUntilFinished();//等待直到数据结束 
    		if(trans.getErrors()>0){ 
    			System.out.println("transformation error"); 
    		}else{ 
    			System.out.println("transformation successfully"); 
    		}
    	} catch (KettleException e) {
    		e.printStackTrace();
    	} 
    } 
    /**
     * 根据文件库路径及名称运行流程
    * @Title: runTransWithFileAbsolutePath 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param fileAbsolutePath
    * @param @param transName    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public   void runTransWithFileAbsolutePath(String fileAbsolutePath,String transName,String[] params,Map<String,String> maps)  { 
    	try {
    		if(filerepository.getRepositoryMeta()==null) {
    			KettleFileRepositoryMeta meta = new KettleFileRepositoryMeta();
    			meta.setBaseDirectory(fileAbsolutePath);
    			meta.setDescription("default");
    			meta.setName("default");
    			meta.setReadOnly(false);
    			meta.setHidingHiddenFiles(true);
    			filerepository.init(meta);
    			
    		}
			
			TransMeta transMeta = filerepository.loadTransformation(new ObjectId() {
				@Override
				public String getId() {
					// TODO Auto-generated method stub
					return transName;
				}
			}, null);
			
			execuTrans(transMeta, params, maps);
    	} catch (KettleException e) {
    		e.printStackTrace();
    	} 
    } 
    /**
     * 根据文件库路径及名称运行流程
     * @Title: runTransWithFileAbsolutePath 
     * @Description: TODO(这里用一句话描述这个方法的作用) 
     * @param @param fileAbsolutePath
     * @param @param transName    设定文件 
     * @return void    返回类型 
     * @throws
     */
    public   void runJobWithFileAbsolutePath(String fileAbsolutePath,String jobName,Map<String,String> maps)  { 
    	try {
    		if(filerepository.getRepositoryMeta()==null) {
    			KettleFileRepositoryMeta meta = new KettleFileRepositoryMeta();
    			meta.setBaseDirectory(fileAbsolutePath);
    			meta.setDescription("default");
    			meta.setName("default");
    			meta.setReadOnly(false);
    			meta.setHidingHiddenFiles(true);
    			filerepository.init(meta);
    			
    		}
    		JobMeta jobMeta  = filerepository.loadJob(new ObjectId() {
    			@Override
    			public String getId() {
    				// TODO Auto-generated method stub
    				return jobName;
    			}
    		}, null);
    		
    		execuJob(jobMeta, maps);
    	} catch (KettleException e) {
    		e.printStackTrace();
    	} 
    } 
    /**
     * 执行转换
    * @Title: execuTrans 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param trans
    * @param @param params
    * @param @param maps    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public void execuTrans(TransMeta transMeta,String[] params,Map<String,String> maps) {
		try {
			Trans trans=new Trans(transMeta); 
			if (maps != null) {
				for (Entry<String, String> entry : maps.entrySet()) {
					String key = entry.getKey();
					String value = entry.getValue();
					trans.setVariable(key, value);
				}
			}
			trans.execute(params);
			trans.waitUntilFinished();// 等待直到数据结束
			if (trans.getErrors() > 0) {
				System.out.println("transformation error");
			} else {
				System.out.println("transformation successfully");
			}

		} catch (KettleException e) {
			e.printStackTrace();
		}
    	
    }
    /**
     * 执行作业job
    * @Title: execuJob 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param jobMeta
    * @param @param maps    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public void execuJob(JobMeta jobMeta,Map<String,String> maps) {
		try {  
    		Job job = new Job(null, jobMeta);  
    		// 向Job 脚本传递参数，脚本中获取参数值：${参数名}  
    		// job.setVariable(paraname, paravalue);  
    		if (maps != null) {
				Set<Entry<String, String>> set = maps.entrySet();
				for (Iterator<Entry<String, String>> it = set.iterator(); it.hasNext();) {
					Entry<String, String> ent = it.next();
					job.setVariable(ent.getKey(), ent.getValue());
				}
			}
    		job.start();  
    		job.waitUntilFinished();  
    		if (job.getErrors() > 0) {  
    			throw new Exception("执行job发生异常");  
    		}  
    	} catch (Exception e) {  
    		e.printStackTrace();  
    	}  
    	
    }
	 /**
	  * 调用资源库执行转换,可选params参数
	 * @Title: runTransWithDb 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param @param databaseMeta
	 * @param @param transName
	 * @param @param params    设定文件 
	 * @return void    返回类型 
	 * @throws
	  */
    public   void runTransWithDb(DatabaseMeta databaseMeta,String transName,String[] params,Map<String,String> maps)  { 
    	try {
    		//选择资源库 
    		KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("kettle","kettle","Transformation description",databaseMeta); 
    		dbrepository.init(kettleDatabaseRepositoryMeta); 
    		//连接资源库 
    		dbrepository.connect("admin","admin"); 
    		RepositoryDirectoryInterface directoryInterface=dbrepository.loadRepositoryDirectoryTree(); 
    		//选择转换 
    		TransMeta transMeta=dbrepository.loadTransformation(transName,directoryInterface,null,true,null);
    		Trans trans=new Trans(transMeta); 
    		if(maps!=null){
    			for(Entry<String, String> entry:maps.entrySet()){
    				String key=entry.getKey();
    				String value=entry.getValue();
    				trans.setVariable(key, value);
    			}
    		}
    		trans.execute(params);
    		trans.waitUntilFinished();//等待直到数据结束 
    		if(trans.getErrors()>0){ 
    			System.out.println("transformation error"); 
    		}else{ 
    			System.out.println("transformation successfully"); 
    		}
    	} catch (KettleException e) {
    		e.printStackTrace();
    	} 
    } 
    
    /**
     * 调用资源库执行转换,可选params参数
     * @Title: runTransWithDb 
     * @Description: TODO(这里用一句话描述这个方法的作用) 
     * @param @param databaseMeta
     * @param @param transName
     * @param @param params    设定文件 
     * @return void    返回类型 
     * @throws
     */
    public   void runTransWithDb(DatabaseMeta databaseMeta,String dir,String transName,String[] params,Map<String,String> maps)  { 
    	try {
//    		 repository=new KettleDatabaseRepository();
    		//选择资源库 
    		KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("kettle","kettle","Transformation description",databaseMeta); 
    		dbrepository.init(kettleDatabaseRepositoryMeta); 
    		dbrepository.disconnect();
    		//连接资源库
    		dbrepository.connect("admin","admin"); 
/*    		if(!repository.isConnected()) {
    			repository.connect("admin","admin"); 
    		}
    		if(repository==null) {
    			 repository=new KettleDatabaseRepository(); 
    		}*/
    		RepositoryDirectoryInterface directoryInterface=dbrepository.findDirectory(dir); 
    		//选择转换 
    		TransMeta transMeta=dbrepository.loadTransformation(transName,directoryInterface,null,true,null);
    		Trans trans=new Trans(transMeta); 
    		if(maps!=null){
    			for(Entry<String, String> entry:maps.entrySet()){
    				String key=entry.getKey();
    				String value=entry.getValue();
    				trans.setVariable(key, value);
    			}
    		}
    		trans.execute(params);
    		trans.waitUntilFinished();//等待直到数据结束 
    		if(trans.getErrors()>0){ 
    			System.out.println("transformation error"); 
    		}else{ 
    			System.out.println("transformation successfully"); 
    		}
    	} catch (KettleException e) {
    		e.printStackTrace();
    	} finally {
    		dbrepository.disconnect();
    	}
    } 
    
    
    /**
     * 调用资源库执行job,可选maps参数
    * @Title: runJobWithDb 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param databaseMeta
    * @param @param jobName
    * @param @param maps    设定文件 
    * @return void    返回类型 
    * @throws
     */
    public  void runJobWithDb(DatabaseMeta databaseMeta,String jobName,Map<String,String> maps)  { 
    	try {
    		//选择资源库 
    		KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("kettle","kettle","Transformation description",databaseMeta); 
    		dbrepository.init(kettleDatabaseRepositoryMeta); 
    		//连接资源库 
    		dbrepository.connect("admin","admin"); 
    		RepositoryDirectoryInterface directoryInterface=dbrepository.loadRepositoryDirectoryTree(); 
    		//选择job 
            JobMeta jobMeta = dbrepository.loadJob(jobName, directoryInterface, null, null);
            Job job = new Job(dbrepository, jobMeta);  
            // 向Job 脚本传递参数，脚本中获取参数值：${参数名}  
            // job.setVariable(paraname, paravalue); 
            if(maps!=null){
            	Set<Entry<String, String>> set=maps.entrySet(); 
            	for(Iterator<Entry<String, String>> it=set.iterator();it.hasNext();){ 
            		Entry<String, String> ent=it.next(); 
            		job.setVariable(ent.getKey(), ent.getValue());  
            	} 
            }
            job.start();  
            job.waitUntilFinished();  
            if (job.getErrors() > 0) {  
            	System.out.println("执行job发生异常");  
            } else{
            	System.out.println("执行job successfully"); 
            }
    	} catch (KettleException e) {
    		e.printStackTrace();
    	} 
    } 
    
    /**
     * 将一个字符串转化为输入流
    * @Title: getStringStream 
    * @Description: TODO(这里用一句话描述这个方法的作用) 
    * @param @param sInputString
    * @param @return    设定文件 
    * @return InputStream    返回类型 
    * @throws
     */
	public static InputStream getStringStream(String sInputString) {
		if (sInputString != null && !sInputString.trim().equals("")) {
			try {
				ByteArrayInputStream tInputStringStream = new ByteArrayInputStream(sInputString.getBytes());
				return tInputStringStream;
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}
		return null;
	}
	public static String getStringFile(String path) {
		 BufferedReader reader = null;  
	        String laststr = "";  
	        try {  
	            FileInputStream fileInputStream = new FileInputStream(path);  
	            InputStreamReader inputStreamReader = new InputStreamReader(  
	                    fileInputStream, "utf-8");  
	            reader = new BufferedReader(inputStreamReader);  
	            String tempString = null;  
	            while ((tempString = reader.readLine()) != null) {  
	                laststr += tempString;  
	            }  
	            reader.close();  
	        } catch (IOException e) {  
	            e.printStackTrace();  
	        } finally {  
	            if (reader != null) {  
	                try {  
	                    reader.close();  
	                } catch (IOException e) {  
	                    e.printStackTrace();  
	                }  
	            }  
	        }  
		return laststr;
	}
  
/*    public static void main(String[] args) throws KettleException {  
        String jobPath = "E:\\kettle-lib\\copymanytablejob.kjb";  
        runJob(jobPath);  
         String ktrPath="E:/kettle-lib/test.ktr";  
//         runTrans(ktrPath);  
//         runTrans(null, ktrPath);
//         runWithDb();
//         KettleEnvironment.init();
         EnvUtil.environmentInit();
//         DatabaseMeta databaseMeta=new DatabaseMeta("kettle","mysql","jdbc","192.168.2.71","kettle","3306","user","123456"); 
         DatabaseMeta databaseMeta=KettleExecuUtil.getDatabaseMeta("192.168.2.71", "3306", "kettle", "user", "123456"); 
         String transName="测试数据异构同步和更新";
         String jobName="测试数据异构同步和更新job";
         KettleExecuUtil.getInstance().runTransWithDb(databaseMeta, transName,null,null);
//         runJobWithDb(databaseMeta, jobName, null);
         
    }  */
	public static void main(String[] args) {
		String str=getStringFile("D:\\患者测试2.ktr");
		InputStream in=getStringStream(str);
		KettleExecuUtil.getInstance().runTrans(null, in);
		System.out.println("very good!!!");
//		KettleExecuUtil.getInstance().runTrans("C:\\Users\\Administrator\\Desktop\\新建文件夹\\患者测试.ktr");
		
//		KettleExecuUtil.getInstance().runTrans("D:\\患者测试2.ktr");
		
	}
	/** 
	* @return dbrepository 
	*/
	public static KettleDatabaseRepository getDbrepository() {
		return dbrepository;
	}
	/** 
	 * @return filerepository 
	 */
	public static KettleFileRepository getFilerepository() {
		return filerepository;
	}
	/** 
	 * 根据参数获取新的资源库
	 * @return dbrepository 
	 */
	public static KettleDatabaseRepository getNewDbrepository(DatabaseMeta databaseMeta) {
		KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("kettle","kettle","Transformation description",databaseMeta); 
		KettleDatabaseRepository kdr=new KettleDatabaseRepository(); 
		kdr.init(kettleDatabaseRepositoryMeta); 
		return kdr;
	}
	/** 
	 * 根据参数获取新的文件库
	 * @return filerepository 
	 */
	public static KettleFileRepository getNewFilerepository(String fileAbsolutePath) {
		KettleFileRepositoryMeta meta = new KettleFileRepositoryMeta();
		meta.setBaseDirectory(fileAbsolutePath);
		meta.setDescription("default");
		meta.setName("default");
		meta.setReadOnly(false);
		meta.setHidingHiddenFiles(true);
		
		KettleFileRepository kfr=new KettleFileRepository();
		kfr.init(meta);
		return kfr;
	}
	/** 
	* @param dbrepository 要设置的 dbrepository 
	*/
	public static void setDbrepository(KettleDatabaseRepository dbrepository) {
		KettleExecuUtil.dbrepository = dbrepository;
	}
	/** 
	* @param filerepository 要设置的 filerepository 
	*/
	public static void setFilerepository(KettleFileRepository filerepository) {
		KettleExecuUtil.filerepository = filerepository;
	}
	
	
    
  
  
}  
